package com.doit.doit47;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.sql.*;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;

public class _02_精准一次性消费 {
    public static final String GROUP_ID = "jia_lin";
    public static void main(String[] args) throws SQLException {
        //消费kafka中的数据
        KafkaConsumer<String, String> consumer = MyKafkaUtils.getConsumer(GROUP_ID, false, "earliest");

        //创建mysql的jdbc连接对象
        Connection conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test01", "root", "123456");
        conn.setAutoCommit(false);
        PreparedStatement user_pps = conn.prepareStatement("insert into user_info values(?,?,?,?)");
        PreparedStatement off_pps = conn.prepareStatement("insert into consumer_offsets values(?,?,?,?) on DUPLICATE key update `offset` = ?");
        //写一个sql去mysql数据库中获取组id和topic对应的partition还有offset
        PreparedStatement get_par_off = conn.prepareStatement("select `partition`,`offset` from consumer_offsets where group_id = ? and topic = ?");


        get_par_off.setString(1,GROUP_ID);
        get_par_off.setString(2,"user");
        ResultSet resultSet = get_par_off.executeQuery();
        ArrayList<TopicPartition> list = new ArrayList<>();
        ArrayList<Long> offsets = new ArrayList<>();
        while (resultSet.next()){
            int partition = resultSet.getInt("partition");
            long offset = resultSet.getLong("offset");
            TopicPartition topicPartition = new TopicPartition("user", partition);
            list.add(topicPartition);
            offsets.add(offset);
        }
        //订阅主题
//        consumer.subscribe(Arrays.asList("user"));

        consumer.assign(list);
        for (int i = 0; i < list.size(); i++) {
            consumer.seek(list.get(i),offsets.get(i));
        }






        while (true){
            ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Long.MAX_VALUE));
            for (ConsumerRecord<String, String> record : poll) {
                try {
                    //9,zss,18,male
                    String value = record.value();
                    String[] arr = value.split(",");
                    //将数据写入到mysql数据库
                    user_pps.setInt(1,Integer.parseInt(arr[0]));
                    user_pps.setInt(3,Integer.parseInt(arr[2]));
                    user_pps.setString(2,arr[1]);
                    user_pps.setString(4,arr[3]);
                    //执行往user_info表中插入数据
                    user_pps.execute();

                    if(arr[0].equals("2")){
                        throw new Exception("你大爷的");
                    }

                    //我还要自己提交偏移量  group_id  topic partition offset 写入到consumer_offsets这张表里面去
                    off_pps.setString(1,GROUP_ID);
                    off_pps.setString(2,record.topic());
                    off_pps.setInt(3,record.partition());
                    off_pps.setLong(4,record.offset()+1);
                    off_pps.setLong(5,record.offset()+1);
                    off_pps.execute();
                    //绑定成一个事务
                    //提交事务
                    conn.commit();
                } catch (Exception e) {
                    //回滚事务
                    conn.rollback();
                    e.printStackTrace();
                }

            }

        }
    }
}
